﻿package net.wangit.etl;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;

import net.wangit.jdbc.DBImplementor;
import net.wangit.jdbc.QueryParameters;
import net.wangit.jdbc.database.Database;
import net.wangit.jdbc.sqltype.SqlType;
import net.wangit.jdbc.sqltype.SqlTypeFactory;
import net.wangit.util.XMLUtil;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.dom4j.Document;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;


public class Extractor {

	private static final Log log = LogFactory.getLog(Extractor.class);
	private String[] names;
	private SqlType[] sqlTypes;

	public Extractor() {
	}

	public int extract(DBImplementor db, String sql, QueryParameters qp,
			String pathPrefix, String tableName) throws Exception {
		ResultSet rs = null;
		int i;
		try {
			rs = db.getResultSet(sql, qp);
			Database database = db.getJDBCSettings().getDatabase();
			Document document = DocumentHelper.createDocument();
			Element root = document.addElement("table").addAttribute("name",
					tableName);
			addColumns(root.addElement("columns"), rs, database);
			Element rows = root.addElement("rows");
			int fileCount = 1;
			int rowCount = 0;
			for (int count = 0; (count = addRows(rows, rs)) > 0;) {
				String path = ETLUtil.getPath(pathPrefix, tableName, fileCount);
				XMLUtil.writeUTF8(document, path);
				fileCount++;
				rowCount += count;
				root.remove(rows);
				rows = root.addElement("rows");
			}

			i = rowCount;
		} catch (Exception e) {
			log.error(e);
			throw e;
		} finally {
			rs.close();
		}
		rs.close();
		return i;
	}

	private void addColumns(Element columns, ResultSet rs, Database database)
			throws Exception {
		ResultSetMetaData meta = rs.getMetaData();
		int columnCount = meta.getColumnCount();
		names = new String[columnCount];
		sqlTypes = new SqlType[columnCount];
		for (int i = 1; i <= columnCount; i++) {
			String columnName = meta.getColumnName(i).toLowerCase();
			int sqlTypeCode = meta.getColumnType(i);
			String columnTypeName = meta.getColumnTypeName(i);
			int precision = meta.getPrecision(i);
			int scale = meta.getScale(i);
			int type = database.getTypeCode(sqlTypeCode, columnTypeName,
					precision, scale);
			Element column = columns.addElement("column");
			column.addAttribute("name", columnName);
			column.addAttribute("type", Integer.toString(type));
			names[i - 1] = columnName;
			sqlTypes[i - 1] = SqlTypeFactory.getSqlType(sqlTypeCode);
		}

	}

	private int addRows(Element rows, ResultSet rs) throws Exception {
		int maxSize = ETLSettings.getMaxSize();
		int count;
		for (count = 0; count < maxSize && rs.next(); count++) {
			Element row = rows.addElement("row");
			for (int i = 0; i < sqlTypes.length; i++) {
				String value = sqlTypes[i].get(rs, names[i]);
				row.addElement("value").addText(ETLUtil.toXmlValue(value));
			}

		}

		return count;
	}

}
